package com.sl.core.engine;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.IoUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.*;
import com.google.common.collect.Maps;
import com.sl.core.engine.boot.ProcessBoot;
import com.sl.core.engine.check.ProcessCompCheckResult;
import com.sl.core.engine.meta.row.ProcessRowDescriptor;
import com.sl.core.engine.meta.trans.ProcessTransDescriptor;
import com.sl.core.engine.plugin.ProcessPluginRegistry;
import com.sl.core.engine.trans.ProcessInstance;
import com.sl.core.engine.trans.ProcessInstancePool;
import lombok.SneakyThrows;
import org.apache.commons.compress.utils.Lists;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
import org.apache.hop.core.ICheckResult;
import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.variables.Variables;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.transform.TransformMeta;
import org.apache.hop.pipeline.transform.TransformMetaDataCombi;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

public class ProcessEngine {
    public static ProcessEngine INSTANCE = new ProcessEngine();


    @SneakyThrows
    public void init() {
        ProcessBoot.init();
    }


    @SneakyThrows
    public Object execute(InputStream inputStream, Object param, Map<String, Object> variableMap) {
        //获取transMeta 可缓存
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream, variableMap);

        ProcessConfig processConfig = new ProcessConfig();
        processConfig.setTid(MapUtil.getStr(variableMap, "tid"));
        processConfig.setOpenId(MapUtil.getStr(variableMap, "openId"));
        return this.execute(transMeta, param, variableMap, processConfig, null, false, null);
    }


    @SneakyThrows
    public Object execute(Object param, ProcessConfig processConfig) {
        return this.execute(param, processConfig, null);
    }

    public Object compensate(ProcessConfig processConfig) {
        processConfig.setCompensateFlag(true);
        return this.execute(null, processConfig, null, false);
    }

    @SneakyThrows
    public Object execute(Object param, ProcessConfig processConfig, Object parent) {
        return execute(param, processConfig, parent, false);
    }

    public Object execute(Object param, ProcessConfig processConfig, Object parent, Boolean asyncFlag) {
        return this.execute(param, processConfig, parent, asyncFlag, null);
    }
    @SneakyThrows
    public Object execute(Object param, ProcessConfig processConfig, Object parent, Boolean asyncFlag, BiFunction<ProcessTransDescriptor,ProcessConfig,ProcessInstance> consumer ) {
        return execute(param,processConfig,parent,asyncFlag,consumer,Maps.newHashMap());
    }
    @SneakyThrows
    public Object execute(Object param, ProcessConfig processConfig, Object parent, Boolean asyncFlag, BiFunction<ProcessTransDescriptor,ProcessConfig,ProcessInstance> consumer,Map<String, Object> variableMap ) {

        String xml = processConfig.getXml();

        if (ObjectUtil.isNull(asyncFlag)) {
            asyncFlag = Boolean.FALSE;
        }

        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(IoUtil.toStream(xml, Charset.defaultCharset())
                , new HashMap<>(), processConfig);
        //初始化一把
        return this.execute(transMeta, param, variableMap, processConfig, parent, asyncFlag, consumer);
    }


    public Object execute(ProcessTransDescriptor transMeta, Object param, Map<String, Object> variableMap, ProcessConfig processConfig, Object parent, Boolean asyncFlag, BiFunction<ProcessTransDescriptor,ProcessConfig,ProcessInstance> consumer) {
        if (StrUtil.isBlank(processConfig.getInstanceId())) {
            processConfig.setInstanceId(IdUtil.fastUUID());
        }
        processConfig.setStartTime(DateUtil.date());


        ProcessInstance trans = null;
        if (ObjectUtil.isNotNull(consumer)) {
            trans=  consumer.apply(transMeta,processConfig);
        } else {
            trans = ProcessInstancePool.INSTANCE.getObj(transMeta, processConfig);
        }

        if (ObjectUtil.isNotNull(parent) && parent instanceof ILoggingObject loggingObject) {
            trans.setParent(loggingObject);
        }

        //变量put
        if (MapUtil.isNotEmpty(variableMap)) {
            for (Map.Entry<String, Object> entry : variableMap.entrySet()) {
                String k = entry.getKey();
                Object v = entry.getValue();
                trans.setVariable(k, StrUtil.toStringOrNull(v));
            }
        }
        Object o = null;
        try {
            if (BooleanUtil.isTrue(asyncFlag)) {
                ProcessInstance finalTrans = trans;
                trans.executeAsync(param, (t) -> {
                    ProcessInstancePool.INSTANCE.returnObj(transMeta, processConfig, finalTrans);
                });
            } else {
                o = trans.executeSync(param, false);
            }
        } finally {
            if (BooleanUtil.isFalse(asyncFlag)) {
                ProcessInstancePool.INSTANCE.returnObj(transMeta, processConfig, trans);
            }

        }
        return o;
    }

    public Object execute(InputStream inputStream, Object param) {
        return this.execute(inputStream, param, new HashMap<>());
    }

    public ProcessPluginRegistry pluginRegistry() {
        return ProcessPluginRegistry.INSTANCE;
    }

    public Boolean stepPause(String stepId, String tid, ProcessConfig processConfig) {
        ProcessTransDescriptor.addStepPause(tid, stepId);
        return true;
    }

    public Boolean transResume(String tid, ProcessConfig processConfig) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor();
        transMeta.stepResume(tid);

        List<ProcessInstance> activeProcessTransByProcessIdAndProcessVersionAndTid = getActiveProcessTransByProcessIdAndProcessVersionAndTid(
                processConfig.getOpenId(), processConfig.getProcessVersion(), tid);
        if (CollUtil.isEmpty(activeProcessTransByProcessIdAndProcessVersionAndTid)) {
            activeProcessTransByProcessIdAndProcessVersionAndTid = getActiveProcessTransByProcessIdAndProcessVersionAndTid(
                    processConfig.getOpenId(), processConfig.getProcessVersion(), null);
        }
        if (CollUtil.isEmpty(activeProcessTransByProcessIdAndProcessVersionAndTid)) {
            return false;
        }
        activeProcessTransByProcessIdAndProcessVersionAndTid.forEach(Pipeline::resumeExecution);
        return true;
    }

    /**
     * 步进
     *
     * @param stepId
     * @param tid
     * @param processConfig
     * @return
     */
    public Boolean stepping(String stepId, String tid, ProcessConfig processConfig) {

        List<ProcessInstance> activeProcessTransByProcessIdAndProcessVersionAndTid = getActiveProcessTransByProcessIdAndProcessVersionAndTid(
                processConfig.getOpenId(), processConfig.getProcessVersion(), tid);
        if (ObjectUtil.isNull(activeProcessTransByProcessIdAndProcessVersionAndTid)) {
            return false;
        }
        for (ProcessInstance processInstance : activeProcessTransByProcessIdAndProcessVersionAndTid) {
            List<TransformMetaDataCombi> steps = processInstance.getTransforms();

            for (TransformMetaDataCombi step : steps) {
                String stepname = step.transform.getTransformName();
                if (StrUtil.equals(stepname, stepId)) {
                    step.transform.resumeRunning();
                }
            }
        }
        return true;
    }

    /**
     * 步进
     *
     * @param tid
     * @param processConfig
     * @return
     */
    public Boolean stepping(String tid, ProcessConfig processConfig) {
        List<ProcessInstance> activeProcessTransByProcessIdAndProcessVersionAndTid = getActiveProcessTransByProcessIdAndProcessVersionAndTid(
                processConfig.getOpenId(), processConfig.getProcessVersion(), tid);
        if (ObjectUtil.isNull(activeProcessTransByProcessIdAndProcessVersionAndTid)) {
            return false;
        }
        activeProcessTransByProcessIdAndProcessVersionAndTid.forEach(Pipeline::resumeExecution);
        return true;
    }

    /**
     * 步进
     *
     * @param stepId
     * @param tid
     * @param processConfig
     * @return
     */
    public Boolean stepResume(String stepId, String tid, ProcessConfig processConfig) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor();
        transMeta.stepResume(tid, stepId);

        List<ProcessInstance> activeProcessTransByProcessIdAndProcessVersionAndTid = getActiveProcessTransByProcessIdAndProcessVersionAndTid(
                processConfig.getOpenId(), processConfig.getProcessVersion(), tid);
        if (ObjectUtil.isNull(activeProcessTransByProcessIdAndProcessVersionAndTid)) {
            return false;
        }
        for (ProcessInstance processInstance : activeProcessTransByProcessIdAndProcessVersionAndTid) {
            List<TransformMetaDataCombi> steps = processInstance.getTransforms();

            for (TransformMetaDataCombi step : steps) {
                String stepname = step.transform.getTransformName();
                if (StrUtil.equals(stepname, stepId)) {
                    step.transform.resumeRunning();
                }
            }
        }
        return true;
    }


    public Boolean transResumeAll(ProcessConfig processConfig) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor();

        List<ProcessInstance> activeProcessTransByProcessIdAndProcessVersionAndTid = getActiveProcessTransByProcessIdAndProcessVersionAndTid(
                processConfig.getOpenId(), processConfig.getProcessVersion(), null
        );
        if (ObjectUtil.isNull(activeProcessTransByProcessIdAndProcessVersionAndTid)) {
            return false;
        }
        activeProcessTransByProcessIdAndProcessVersionAndTid.forEach(v -> {
            transMeta.stepResume(v.getTid());
            v.resumeExecution();
        });
        return true;

    }

    public List<ProcessInstance> getActiveProcessTransByProcessIdAndProcessVersionAndTid(String openId, String processVersion, String tid) {
        List<ProcessInstance> activeProcessTrans = this.getActiveProcessTrans();
        return activeProcessTrans.stream().filter(pi -> {

            if (StrUtil.isNotBlank(openId) && !StrUtil.equals(openId, pi.getOpenId())) {
                return false;
            }
            if (StrUtil.isNotBlank(processVersion) && !StrUtil.equals(processVersion, pi.getProcessVersion())) {
                return false;
            }
            if (StrUtil.isNotBlank(tid)){
                return  true;
            }
            if (StrUtil.isNotBlank(tid) && !StrUtil.equals(tid, pi.getTid())) {
                return false;
            }
            return true;
        }).collect(Collectors.toList());
    }

    public List<ProcessInstance> getActiveProcessTrans() {

        ProcessInstancePool instance = ProcessInstancePool.INSTANCE;
        GenericKeyedObjectPool<ProcessInstancePool.ProcessInstancePoolKey, ProcessInstance> pool = instance.getPool();

        Object fieldValue = ReflectUtil.getFieldValue(pool, "poolMap");
        if (ObjectUtil.isNull(fieldValue)) {
            return null;
        }
        AtomicReference<List<ProcessInstance>> atomicReference = new AtomicReference<>(Lists.newArrayList());
        if (fieldValue instanceof Map fMap) {
            fMap.forEach((k, v) -> {
                Object allObjects = ReflectUtil.getFieldValue(v, "allObjects");
                if (ObjectUtil.isNull(allObjects)) {
                    return;
                }
                if (allObjects instanceof Map allMap) {

                    allMap.forEach((kk, vv) -> {
                        Object processInstanceObj = ReflectUtil.getFieldValue(kk, "instance");
                        if (ObjectUtil.isNull(processInstanceObj)) {
                            return;
                        }
                        if (processInstanceObj instanceof ProcessInstance pi) {
                            atomicReference.get().add(pi);
                        }
                    });
                }

            });
        }
        return atomicReference.get();
    }

    public void clearPool() {
        ProcessInstancePool.INSTANCE.clear();
    }

    @SneakyThrows
    public Map<String, Object> getCurrentInputDefine(ByteArrayInputStream inputStream, String stepCode) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        ProcessRowDescriptor processRowDescriptor = new ProcessRowDescriptor();
        return processRowDescriptor.toInputDefine(transMeta, stepCode);
    }


    @SneakyThrows
    public Map<String, Object> getCurrentOutputDefine(ByteArrayInputStream inputStream, String stepCode) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        ProcessRowDescriptor processRowDescriptor = new ProcessRowDescriptor();
        return processRowDescriptor.toOutputDefine(transMeta, stepCode);
    }

    @SneakyThrows
    public List<String> getNextTransforms(ByteArrayInputStream inputStream, String stepCode) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        //找当前组件是否存在input
        TransformMeta step = transMeta.findTransform(stepCode);

        List<TransformMeta> nextTransforms = transMeta.findNextTransforms(step);
        return nextTransforms.stream().map(TransformMeta::getName).collect(Collectors.toList());
    }

    @SneakyThrows
    public List<ProcessCompCheckResult> check(ByteArrayInputStream inputStream) {
        List<ICheckResult> remarks = Lists.newArrayList();
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        transMeta.checkTransforms(remarks, false, null, new Variables(), null);

        List<ICheckResult> collect = remarks.stream().filter(v -> v.getType() != ICheckResult.TYPE_RESULT_OK && v.getType() != ICheckResult.TYPE_RESULT_WARNING).toList();

        return ProcessCompCheckResult.of(collect);
    }

    @SneakyThrows
    public List<Map<String, Object>> getCurrentInputDefineGroup(ByteArrayInputStream inputStream, String stepCode) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        ProcessRowDescriptor processRowDescriptor = new ProcessRowDescriptor();
        return processRowDescriptor.toInputDefineGroup(transMeta, stepCode);
    }

    @SneakyThrows
    public Map<String, Object> getProcessEventInputDefine(ByteArrayInputStream inputStream) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        ProcessRowDescriptor processRowDescriptor = new ProcessRowDescriptor();
        return processRowDescriptor.toProcessEventInputDefine(transMeta);
    }

    @SneakyThrows
    public Map<String, Object> getProcessEventOutputDefine(ByteArrayInputStream inputStream) {
        ProcessTransDescriptor transMeta = new ProcessTransDescriptor(inputStream);
        ProcessRowDescriptor processRowDescriptor = new ProcessRowDescriptor();
        return processRowDescriptor.getProcessEventOutputDefine(transMeta);
    }


}
